package com.sl.core.engine.step.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.*;
import cn.hutool.json.JSONUtil;
import cn.hutool.script.ScriptUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sl.core.engine.data.impl.AbstractProcessCompContext;
import com.sl.core.engine.datasource.AnylineServiceAdaptor;
import com.sl.core.engine.domain.ProcessRowEvent;
import com.sl.core.engine.domain.ProcessStepPauseConfig;
import com.sl.core.engine.listener.ProcessRowLogListener;
import com.sl.core.engine.listener.RowListener;
import com.sl.core.engine.meta.impl.AbstractProcessCompDescriptor;
import com.sl.core.engine.meta.row.ProcessRowDescriptor;
import com.sl.core.engine.meta.trans.ProcessTransDescriptor;
import com.sl.core.engine.meta.valuemeta.ValueMetaArray;
import com.sl.core.engine.meta.valuemeta.ValueMetaConvert;
import com.sl.core.engine.meta.valuemeta.ValueMetaObject;
import com.sl.core.engine.rowset.ProcessAbstractDbRowSet;
import com.sl.core.engine.step.rowto.ProcessCompPutRowToConfig;
import com.sl.core.engine.trans.ProcessInstance;
import com.sl.core.engine.utils.ProcessTypeUtil;
import com.sl.core.utils.JsonUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.anyline.service.AnylineService;
import org.apache.hop.core.IRowSet;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.engine.IPipelineEngine;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMeta;

import javax.script.Bindings;
import javax.script.ScriptEngine;
import javax.script.SimpleBindings;
import java.lang.reflect.Type;
import java.util.*;

@Slf4j
public abstract class AbstractProcessComp<I, R> extends BaseTransform {

    protected AbstractProcessCompDescriptor<?, ?> meta;


    protected AbstractProcessCompContext data;


    @Getter
    @Setter
    private List<RowListener> rowListenerList = Lists.newArrayList();

    @Getter
    private List<Throwable> throwables = Lists.newArrayList();

    public AbstractProcessComp(AbstractProcessCompDescriptor abstractProcessCompDescriptor) {
        this(abstractProcessCompDescriptor.getCompConstructor().getStepMeta(),
                (AbstractProcessCompDescriptor) abstractProcessCompDescriptor.getCompConstructor().getTransformMeta(),
                (AbstractProcessCompContext) abstractProcessCompDescriptor.getCompConstructor().getStepDataInterface(),
                abstractProcessCompDescriptor.getCompConstructor().getCopyNr(),
                abstractProcessCompDescriptor.getCompConstructor().getTransMeta(),
                abstractProcessCompDescriptor.getCompConstructor().getTrans());
    }


    public AbstractProcessComp(TransformMeta transformMeta, AbstractProcessCompDescriptor meta, AbstractProcessCompContext data, int copyNr, PipelineMeta pipelineMeta, Pipeline pipeline) {
        super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
        this.meta = meta;
        this.data = data;
        rowListenerList.add(ProcessRowLogListener.INSTANCE);
    }

    @Override
    public boolean init() {
//        this.setLogLevel(meta.getLog().getLogLevel());
        throwables.clear();
        return super.init();
    }

    public void addThrowable(Throwable throwable) {
        throwables.add(throwable);
    }

    @Override
    public void dispose() {
        super.dispose();
        //失败了会被清空掉，需要考虑怎么处理
        //throwables.clear();
    }


    public <T> T getDescriptor() {
        return (T) meta;
    }

    @Override
    public ProcessInstance getPipeline() {
        return (ProcessInstance) super.getPipeline();
    }

    public AnylineService getAnyLineService(String dataSourceId, Boolean startTransactionFlag) {
        ProcessInstance pipeline = getPipeline();
        AnylineServiceAdaptor anylineServiceAdaptor = pipeline.getAnylineServiceAdaptor();
        return anylineServiceAdaptor.getAnyLineService(pipeline.getTid(), dataSourceId, startTransactionFlag);
    }


    public AnylineService getAnyLineService(String dataSourceId) {
        ProcessInstance pipeline = getPipeline();
        AnylineServiceAdaptor anylineServiceAdaptor = pipeline.getAnylineServiceAdaptor();
        return anylineServiceAdaptor.getAnyLineService(pipeline.getTid(), dataSourceId);
    }


    @Override
    public String getVariable(String variableName) {
        String variable = super.getVariable(variableName);
        if (StrUtil.isNotBlank(variable)) {
            return variable;
        }
        return this.meta.getStrAttr(variableName);
    }

    @SneakyThrows
    @Override
    public boolean processRow() {
        //输入
        Object input = null;
        //是否结束
        Boolean isEnd = false;
        //数据来源名称
        String originTransformName = null;

        String thisTransformName = getTransformName();

        //判断是不是结束组件
        Boolean isEndStep = meta.getIsEndStep();

        //如果是开始步骤 需要从argument里面取数据
        Boolean isStartStep = meta.getIsStartStep();

        ProcessRowEvent processRowEvent = new ProcessRowEvent();
        processRowEvent.setUid(IdUtil.fastUUID());
        processRowEvent.setStartStepFlag(isStartStep);
        processRowEvent.setFirstFlag(first);
        processRowEvent.setInstanceId(this.getPipeline().getInstanceId());
        processRowEvent.setEndStepFlag(isEndStep);
        processRowEvent.setTid(this.getPipeline().getTid());
//        processRowEvent.setInput(input);
        processRowEvent.setLogChannel(getLogChannel());
        processRowEvent.setOriginTransformName(originTransformName);
        processRowEvent.setThisTransformName(thisTransformName);
        Boolean compensateFlag = this.getPipeline().getCompensateFlag();
        if (ObjectUtil.isNull(compensateFlag)) {
            compensateFlag = false;
        }
        processRowEvent.setCompensateFlag(compensateFlag);

        //不是补偿，并且是中断流程的话，直接中断了
//        if(!compensateFlag && this instanceof AbstractProcessBreakComp){
//            setOutputDone();
//            return false;
//        }
        Boolean breakFlag = this.isBreak(compensateFlag);
        if (BooleanUtil.isTrue(breakFlag)) {
            setOutputDone();
            return false;
        }

        //获取数据来源
        IRowSet inputRowSet = getCurrentInputRowSet();
        if (ObjectUtil.isNotNull(inputRowSet)) {
            originTransformName = inputRowSet.getOriginTransformName();
        }


        //补偿的话，需要把库里的error改成0
        if (BooleanUtil.isTrue(compensateFlag)) {
            if (inputRowSet instanceof ProcessAbstractDbRowSet processAbstractDbRowSet) {
                //dataInput 应该传入这个 ，一批有问题要处理
                processAbstractDbRowSet.errorToNoRead();
            }
        }
        //获取输入
        Object[] row = getRow();
        if (ArrayUtil.isNotEmpty(row)) {
            //转为Map
            IRowMeta inputRowMeta = getInputRowMeta();
            //如果没有定义，则按照这个来
            if (!inputRowMeta.isEmpty()) {
                input = this.rowToMap(row, inputRowMeta);
            } else {
                input = this.rowToMap(row);
            }
        }
        //是开始组件，并且输入为null并且是第一次 则从argment里面读取 并且不是补偿
        if (BooleanUtil.isTrue(isStartStep) && ObjectUtil.isNull(input) && first) {
            input = this.processStart();
            this.incrementLinesRead();
            isEnd = true;
        }
        //如果数据为空 标记最后
        if (ObjectUtil.isNull(input)) {
            isEnd = true;
        }
        processRowEvent.setInput(input);

        if (BooleanUtil.isTrue(first)) {
            for (RowListener rowListener : rowListenerList) {
                rowListener.start(processRowEvent);
            }
        }

        //打印输入
        for (RowListener rowListener : rowListenerList) {
            rowListener.input(processRowEvent);
        }

        //debug 暂停
        stepDebugCondition(input);

        //如果是第一次 改成false
        if (first) {
            first = false;
        }
        //获取组件上面的范型
        Type paramTypeClass = ProcessTypeUtil.getParamTypeClass(this.getClass());
        Class<?> aClass = TypeUtil.getClass(paramTypeClass);
        //入参转换
        inputTransfer(input, originTransformName, paramTypeClass, aClass);
        R r = null;
        //如果分片完成
        if (data.isCut(isEnd)) {
            I dataInput = (I) data.getInput(Collection.class.isAssignableFrom(aClass));

            //设置一把
            processRowEvent.setInput(dataInput);
            //消费
            for (RowListener rowListener : rowListenerList) {
                rowListener.process(processRowEvent);
            }
            try {
                r = this.input(dataInput);

                if (inputRowSet instanceof ProcessAbstractDbRowSet processAbstractDbRowSet) {
                    //dataInput 应该传入这个 ，一批有问题要处理
                    processAbstractDbRowSet.readFinish(dataInput);
                }

            } catch (Throwable e) {
                for (RowListener rowListener : rowListenerList) {
                    rowListener.error(processRowEvent);
                }
//                IRowSet currentInputRowSet = getCurrentInputRowSet();
                if (inputRowSet instanceof ProcessAbstractDbRowSet processAbstractDbRowSet) {
                    //dataInput 应该传入这个 ，一批有问题要处理
                    processAbstractDbRowSet.error(dataInput, e);
                }
                throw e;
            }
            //结果往下游传递
            resultTransferRow(r);
        }

        //如果是结束步骤，则处理一把
        if (BooleanUtil.isTrue(isEndStep)) {
            this.processEnd(r, isEnd);
        }

        for (RowListener rowListener : rowListenerList) {
            processRowEvent.setOutput(r);
            rowListener.output(processRowEvent);
        }

        //如果是没有参数了 ，则步骤结束
        if (BooleanUtil.isTrue(isEnd)) {
            setOutputDone();
            for (RowListener rowListener : rowListenerList) {
                rowListener.end(processRowEvent);
            }
            return false;
        }
        return true;
    }

    protected Boolean isBreak(Boolean compensateFlag) {
        return false;
    }

    private IRowSet getCurrentInputRowSet() {
        IRowSet iRowSet = null;
        try {
            int currentInputRowSetNr = this.getCurrentInputRowSetNr();
            List<IRowSet> inputRowSets = this.getInputRowSets();
            iRowSet = CollUtil.get(inputRowSets, currentInputRowSetNr);
        } catch (Exception e) {

        }
        return iRowSet;
    }

    private void inputTransfer(Object input, String originTransformName, Type paramTypeClass, Class<?> aClass) {

        if (ObjectUtil.isNotNull(input)) {
            if (input instanceof Collection<?> cc) {
                for (Object o : cc) {
                    //获取入参数
                    Object pp = ValueMetaConvert.objToBean(paramTypeClass, o);
                    //如果是一个集合那么需要收集一把数据
                    data.addInput(pp, aClass, originTransformName);
                }
            } else {
                //获取入参数
                Object pp = ValueMetaConvert.objToBean(paramTypeClass, input);
                //如果是一个集合那么需要收集一把数据
                data.addInput(pp, aClass, originTransformName);
            }
        }
    }


    @SneakyThrows
    private void resultTransferRow(Object result) {

        if (ObjectUtil.isNull(result)) {
            return;
        }
        if (result instanceof ProcessCompPutRowToConfig putRowToConfig) {

            List<String> putToTransformNameList = putRowToConfig.getPutToTransformNameList();

            Object data1 = putRowToConfig.getData();
            if (CollUtil.isEmpty(putToTransformNameList)) {
                return;
            }
            for (String s : putToTransformNameList) {
                resultTransferRow(data1, s);
            }

        } else {
            resultTransferRow(result, null);
        }
    }

    @SneakyThrows
    private void resultTransferRow(Object result, String nextTransformName) {

        if (ObjectUtil.isNull(result)) {
            return;
        }

        IRowMeta inputRowMeta = getInputRowMeta();
        RowMeta outputMeta = meta.getResultDescriptor((RowMeta) inputRowMeta);
        ProcessRowDescriptor processRowDescriptor = null;
        if (ObjectUtil.isNotNull(outputMeta)) {
            if (outputMeta instanceof ProcessRowDescriptor rowDescriptor) {
                processRowDescriptor = rowDescriptor;
            } else {
                processRowDescriptor = new ProcessRowDescriptor();
                processRowDescriptor.addRowMeta(outputMeta);
            }
        }

        if (ObjectUtil.isNull(processRowDescriptor) || processRowDescriptor.isEmptyAndNotBasic()) {
            outputMeta = ProcessRowDescriptor.objectToRowMeta(result);
        }


        if (result instanceof Collection) {
            Collection c = (Collection) result;
            for (Object o : c) {
                if (o instanceof Map) {
                    Object[] outRows = mapToRow((Map<String, Object>) o, outputMeta);

                    if (StrUtil.isNotBlank(nextTransformName)) {
                        IRowSet outputRowSet = this.findOutputRowSet(nextTransformName);
                        putRowTo(outputMeta, outRows, outputRowSet);
                    } else {
                        putRow(outputMeta, outRows);
                    }

                } else {
                    Object[] arr = new Object[1];
                    arr[0] = o;
                    if (StrUtil.isNotBlank(nextTransformName)) {
                        IRowSet outputRowSet = this.findOutputRowSet(nextTransformName);
                        putRowTo(outputMeta, arr, outputRowSet);
                    } else {
                        putRow(outputMeta, arr);
                    }
                }
            }
        } else if (result instanceof Map) {
            Object[] outRows = mapToRow((Map<String, Object>) result, outputMeta);

            if (StrUtil.isNotBlank(nextTransformName)) {
                IRowSet outputRowSet = this.findOutputRowSet(nextTransformName);
                putRowTo(outputMeta, outRows, outputRowSet);
            } else {
                putRow(outputMeta, outRows);
            }
        } else {
            Object[] arr = new Object[1];
            arr[0] = result;

            if (StrUtil.isNotBlank(nextTransformName)) {
                IRowSet outputRowSet = this.findOutputRowSet(nextTransformName);
                putRowTo(outputMeta, arr, outputRowSet);
            } else {
                putRow(outputMeta, arr);
            }
        }
    }


    @SneakyThrows
    private void stepDebugCondition(Object input) {

        IPipelineEngine pipeline = getPipeline();

        if (!(pipeline instanceof ProcessInstance processInstance)) {
            return;
        }
        String tid = processInstance.getTid();
        String stepID = this.getTransformName();
        PipelineMeta pipelineMeta = this.getPipelineMeta();
        if (!(pipelineMeta instanceof ProcessTransDescriptor processTransDescriptor)) {
            return;
        }
        ProcessStepPauseConfig stepPause = processTransDescriptor.getStepPause(tid, stepID);
        if (ObjectUtil.isNull(stepPause)) {
            return;
        }
        String condition = stepPause.getCondition();
        if (StrUtil.isNotBlank(condition)) {
            ScriptEngine groovyEngine = ScriptUtil.createGroovyEngine();
            Bindings bindings = new SimpleBindings();
            bindings.put("input", input);
            Object eval = groovyEngine.eval(condition, bindings);
            Boolean result = Convert.convert(Boolean.class, eval);
            if (BooleanUtil.isTrue(result)) {
                pauseRunning();
            }
        } else {
            pauseRunning();
        }
        getPipeline().callStepStatus();
        while (isPaused() && !isStopped()) {
            try {
                Thread.sleep(100L);
            } catch (InterruptedException var14) {
                throw new HopTransformException(var14);
            }
        }
    }


    public R input(I param) {
        return null;
    }

    /**
     * 输入是否是数组
     *
     * @return
     */
    protected Boolean inputIsArray() {
        IRowMeta inputRowMeta = this.getInputRowMeta();
//        String sql = BeanUtil.getStr(meta.getAttributeMap(), "input.");
        Boolean inputIsArray = false;
        if (inputRowMeta instanceof ProcessRowDescriptor processRowDescriptor) {
            inputIsArray = processRowDescriptor.getIsArray();
        }
        inputIsArray = ObjectUtil.defaultIfNull(inputIsArray, false);
        return inputIsArray;
    }

    @SneakyThrows
    public Object processStart() {

        String json = getPipeline().getParameterValue("input");
        //false是不继续了
        if (StrUtil.isBlank(json)) {
            return null;
        }
        //如果顶层是对象 这里是不是不需要做映射
        Object object = null;
        if (JSONUtil.isTypeJSON(json)) {
            object = JSONUtil.parse(json);
        } else {
            object = json;
        }
        return object;
    }


    @SneakyThrows
    public void processEnd(R r, Boolean isEnd) {
        if (ObjectUtil.isNotNull(r)) {

            if (r instanceof Map c) {
                Map cb = new HashMap();
                for (Object key : c.keySet()) {
                    //如果是自动填充的需要处理一把
                    if (StrUtil.startWith(key.toString(), "args", true)) {
                        data.addResult(c.get(key));
                    } else {
                        cb.put(key, c.get(key));
                    }
                }
                if (MapUtil.isNotEmpty(cb)) {
                    data.addResult(cb);
                }
            } else {
                data.addResult(r);
            }

        }

        if (BooleanUtil.isTrue(isEnd)) {
            String str1 = JsonUtil.toStr(data.getResult(false));
            if (StrUtil.isNotBlank(str1)) {
                getPipeline().setParameterValue("output", str1);
            }

        }
    }


    /**
     * 数组形式的入参转为map
     *
     * @param rows
     * @param rowMeta
     * @return
     */
    protected Map<String, Object> rowToMap(Object[] rows, IRowMeta rowMeta) {
        Map<String, Object> hm = new LinkedHashMap<>();
        List<IValueMeta> valueMetaList = rowMeta.getValueMetaList();
        for (int i = 0; i < valueMetaList.size(); i++) {
            IValueMeta iValueMeta = valueMetaList.get(i);
            if (rows.length - 1 < i) {
                hm.put(iValueMeta.getName(), null);
            } else {
                hm.put(iValueMeta.getName(), rows[i]);
            }
        }
        Object row = rows[rows.length - 1];
        if (ProcessAbstractDbRowSet.isHIndex(row)) {
            hm.put("hIndex", row);
        }
        return hm;
    }


    public Map<String, Object> rowToMap(Object[] tr) {
        Map<String, Object> hashMap = Maps.newLinkedHashMap();

        int i = 0;
        for (Object o : tr) {
            hashMap.put("args" + i, o);
            i++;
        }

        return hashMap;
    }


    protected Object[] mapToRow(Map<String, Object> value, IRowMeta rowMeta) {
        Object[] objects = new Object[rowMeta.size()];
        if (MapUtil.isEmpty(value)) {
            return objects;
        }
        for (Map.Entry<String, Object> entry : value.entrySet()) {
            int outIndex = rowMeta.indexOfValue(entry.getKey());
            if (outIndex < 0) {
                continue;
            }
            objects[outIndex] = entry.getValue();
        }
        //校验值是否都是空的，如果是则直接返回
        long count = Arrays.stream(objects).filter(ObjectUtil::isNull).count();
        if (count == objects.length) {
            return new Object[0];
        }
        return objects;
    }


    @SneakyThrows
    /**
     * 单行输入值类型转换
     */ protected static Object[] jsonObjectTransferByRowDefine(Object json, IRowMeta sourceRowMeta) {
        //需要转换的
        IRowMeta targetRowMeta = sourceRowMeta.cloneToType(IValueMeta.TYPE_STRING);
        Map<?, ?> map = (Map<?, ?>) json;
        return mapTransferByRowDefine(sourceRowMeta, targetRowMeta, map);
    }

    @SneakyThrows

    protected static Object[] mapTransferByRowDefine(IRowMeta rowMeta, IRowMeta coverMeta, Map map) {
        //获取值列表
        List<IValueMeta> list = rowMeta.getValueMetaList();

        Object[] data = new Object[list.size()];

        for (int i = 0; i < list.size(); i++) {

            IValueMeta valueMeta = list.get(i);

            Object value = map.get(valueMeta.getName());

            //如果是数组 或者对象 或者值为空
            if (valueMeta.getType() == Integer.parseInt(ValueMetaObject.TYPE) || value == null || valueMeta.getType() == Integer.parseInt(ValueMetaArray.TYPE)) {
                //此处是特殊处理,JSON输入可能包含子对象，按照当前逻辑，子对象需要传入到下一个步骤 JSON分割组件处理，所以针对子对象数据不做数据转换处理；
                //tips: JSON分割组件需也需要做值转换
                data[i] = value;
            } else {
                IValueMeta convertMeta = coverMeta.getValueMeta(i);
                convertMeta.setStorageType(IValueMeta.STORAGE_TYPE_NORMAL);

                if (value instanceof byte[]) {
                    value = valueMeta.convertBinaryStringToNativeType((byte[]) value);
                }
//                else if (value instanceof CLOB) {
//                    value = valueMeta.convertBinaryStringToNativeType(((CLOB) value).getBytes());
//                }
                //在转下
                data[i] = valueMeta.convertDataFromString(value.toString(), convertMeta, null, null, 0);
            }
        }
        return data;
    }


    @Override
    public Object[] getRow() throws HopException {
        //todo logfile
        return super.getRow();
    }

    @Override
    public void putRowTo(IRowMeta rowMeta, Object[] row, IRowSet rowSet)
            throws HopTransformException {
        //需要处理下
        //todo logfile
        super.putRowTo(rowMeta, row, rowSet);
    }

    @Override
    public void putRow(IRowMeta rowMeta, Object[] row) throws HopTransformException {
        //todo logfile
        super.putRow(rowMeta, row);
    }

    @SneakyThrows
    protected List<Object[]> jsonArrayTransferByRowDefine(Object json, IRowMeta rowMeta) {
        IRowMeta coverMeta = rowMeta.cloneToType(IValueMeta.TYPE_STRING);

        List<Object[]> result = new ArrayList<>();
        List<Map> maps = (List<Map>) json;

        if (maps != null) {
            for (Map map : maps) {
                result.add(mapTransferByRowDefine(rowMeta, coverMeta, map));
            }
        }
        return result;
    }
}
